package com.iakuil.app.common.redis.utils;

import com.iakuil.app.common.core.util.SpringUtils;
import lombok.experimental.UtilityClass;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.lang.Nullable;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Redis Stream命令工具类
 *
 * <p>Redis版本必须在5.0及以上。
 *
 * @author Kai
 */
@UtilityClass
public class StreamUtils {

    private final StringRedisTemplate redisTemplate = SpringUtils.getBean("stringRedisTemplate");

    /**
     * 获取Stream Key信息
     *
     * @param key Stream Key
     * @return Key信息
     */
    @Nullable
    public StreamInfo.XInfoStream info(String key) {
        return Boolean.TRUE.equals(redisTemplate.hasKey(key)) ? redisTemplate.opsForStream().info(key) : null;
    }

    /**
     * 消费组信息
     *
     * @param key Stream Key
     * @return 消费组信息
     */
    public StreamInfo.XInfoGroups groups(String key) {
        return redisTemplate.opsForStream().groups(key);
    }

    /**
     * 创建消费组
     *
     * @param key   Stream Key
     * @param group 消费组
     * @return 成功返回OK，否则返回null
     */
    public String createGroup(String key, String group) {
        return redisTemplate.opsForStream().createGroup(key, group);
    }

    /**
     * 消费组信息
     *
     * @param key   Stream Key
     * @param group 消费组
     * @return 消费组信息
     */
    public StreamInfo.XInfoConsumers consumers(String key, String group) {
        return redisTemplate.opsForStream().consumers(key, group);
    }

    /**
     * 确认已消费
     *
     * @param key       Stream Key
     * @param group     消费组
     * @param recordIds 需要确认的消息ID列表
     * @return 消息ID
     */
    public Long ack(String key, String group, String... recordIds) {
        return redisTemplate.opsForStream().acknowledge(key, group, recordIds);
    }

    /**
     * 追加消息
     *
     * @param key   Stream Key
     * @param field 属性名
     * @param value 属性值
     * @return 消息ID
     */
    public String add(String key, String field, Object value) {
        Map<String, Object> content = new HashMap<>(1);
        content.put(field, value);
        return add(key, content);
    }

    @Nullable
    public String add(String key, Map<String, Object> content) {
        RecordId id = redisTemplate.opsForStream().add(key, content);
        return id == null ? null : id.getValue();
    }

    /**
     * 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
     * 消息存储在stream的节点下，删除时仅对消息做删除标记，当一个节点下的所有条目都被标记为删除时，销毁节点
     *
     * @param key       Stream Key
     * @param recordIds 需要确认的消息ID列表
     * @return 被移除的数量
     */
    public Long del(String key, String... recordIds) {
        return redisTemplate.opsForStream().delete(key, recordIds);
    }

    /**
     * 消息长度
     *
     * @param key Stream Key
     * @return 消息数量
     */
    public Long len(String key) {
        return redisTemplate.opsForStream().size(key);
    }

    /**
     * 从开始读
     *
     * @param key Stream Key
     * @return 消息列表
     */
    @SuppressWarnings("unchecked")
    public List<MapRecord<String, Object, Object>> read(String key) {
        return redisTemplate.opsForStream().read(StreamOffset.fromStart(key));
    }

    /**
     * 消息裁剪
     *
     * @param key   Stream Key
     * @param count Stream长度
     * @return 消息数量
     */
    public Long trim(String key, long count) {
        return redisTemplate.opsForStream().trim(key, count);
    }

    /**
     * 从指定的ID开始读
     *
     * @param key      Stream Key
     * @param recordId 需要确认的消息ID
     * @return 消息列表
     */
    @SuppressWarnings("unchecked")
    public List<MapRecord<String, Object, Object>> read(String key, String recordId) {
        return redisTemplate.opsForStream().read(StreamOffset.from(MapRecord.create(key, new HashMap<>(1)).withId(RecordId.of(recordId))));
    }
}
